package io.kimmking.common.rocketmq.base;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONValidator;
import io.kimmking.dubbo.common.unit.ThrowableActionV2;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public abstract class BaseIConsumer<T> implements MessageListenerConcurrently {

    protected T body;

    /**
     * 校验消费的消息体
     * @param message
     * @param clazz
     * @return
     */
    protected boolean isVerify(MessageExt message, final Class<T> clazz) {
        String body = new String(message.getBody());
        if (StringUtils.isBlank(body)) {
            log.info("MQ消息体为空");
            return false;
        }

        // log.info("MQ消费者 - 接收到的消息内容：{}", body);
        ThrowableActionV2.execute(() -> {
            JSONValidator a = JSONValidator.from(body);
            if (a.validate()) {
                this.body = JSON.parseObject(body, clazz);
            } else {
                this.body = (T) body;
            }
        });
        return true;
    }

    /**
     * 对MQ消息的拦截处理
     */
    protected abstract void handle(List<MessageExt> msgs);
}
